package com.maimao.search.listener;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.maimao.core.model.resp.Result;
import com.maimao.message.kafka.sender.KafkaSender;
import com.maimao.model.goods.vo.GoodsInfo;
import com.maimao.search.client.GoodsClient;
import com.maimao.search.model.goods.EsGoods;
import com.maimao.search.service.DocService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author MaoLin Wang
 * @date 2021/1/3 11:40 上午
 */
@Component
public class GoodsListener {

    private Logger logger = LoggerFactory.getLogger(GoodsListener.class);

    @Autowired
    private KafkaSender kafkaSender;
    @Autowired
    private GoodsClient goodsClient;
    @Autowired
    private DocService docService;

    @Value("${maimao.es.goods.index}")
    private String index;
    @Value("${maimao.es.goods.type}")
    private String type;

    /**
     * 监听发布成功的商品 同步到ES
     *
     * @param record
     * @param ack
     */
    @KafkaListener(topics = {"goods-public"}, containerFactory = "factoryAck")
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        String value = "";
        try {
            value = record.value().toString();
            if (StringUtils.isEmpty(value)) {
                logger.error("商品发布消息接收异常{}", record);
                return;
            }
            logger.info("收到商品发布消息: " + record.value().toString());

            Long goodsId = Long.valueOf(value);
            Result<GoodsInfo> goodsInfoResult =
                    goodsClient.goodsInfoPageByGoodsId(goodsId);
            System.out.println(goodsInfoResult.getData());
            if (goodsInfoResult != null && goodsInfoResult.getData() != null) {
                GoodsInfo goodsInfo = goodsInfoResult.getData();
                List<EsGoods> esGoods = docService.goodsInfo2EsGoods(Lists.newArrayList(goodsInfo));
                if (esGoods != null && esGoods.size() > 0) {
                    esGoods.stream().forEach(goods -> {
                        docService.createDoc(index, goods.getId(), JSON.toJSONString(goods), type);
                    });
                }

            }
        } finally {
            logger.info("商品发布消息确认：{}", value);
            ack.acknowledge();
        }
    }
}
